package com.yb.guigu.pick.activity.listener.redis;

import com.yb.guigu.pick.activity.channel.IRedisChannel;
import com.yb.guigu.pick.activity.channel.RedisChannel;
import com.yb.guigu.pick.activity.interceptor.IRedisStreamInterceptor;
import com.yb.guigu.pick.activity.interceptor.RedisStreamInterceptor;
import lombok.Data;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Component
@Data
public abstract class RedisStreamConsumer<T extends RedisStreamConsumer<?>> implements  FactoryBean<T> , StreamListener<String, MapRecord<String,String,String>>, ApplicationContextAware {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private String group;

    private String consumer;


    private boolean autoAck;


    private ApplicationContext applicationContext;

    @Autowired
    private List<IRedisStreamInterceptor> redisStreamInterceptors;

    private  Class<? extends RedisStreamConsumer<?>> type;

    public RedisStreamConsumer(){
        //在这里 this 代表子类对象
        //获取父类的泛型<T>
        Type genericSuperclass = this.getClass().getGenericSuperclass();
        //Type强转成子类ParameterizedType这样就可以通过子类的getActualTypeArguments获取类对象
        ParameterizedType parameterizedType = (ParameterizedType)genericSuperclass;
        type = (Class<? extends RedisStreamConsumer<?>>)parameterizedType.getActualTypeArguments()[0];
    }
    @Override
     public final void onMessage(MapRecord<String, String, String> message){
        String stream = message.getStream();
        RecordId id = message.getId();
        RedisChannel redisChannel=new RedisChannel();
        redisChannel.setConsumer(consumer);
        redisChannel.setStreamKey(stream);
        redisChannel.setGroup(group);
        redisChannel.setId(id);
        redisChannel.setStringRedisTemplate(stringRedisTemplate);
        redisChannel.setAutoAck(autoAck);

        handleMessage(redisChannel,message);

    }

    private void handleMessage(IRedisChannel redisChannel,MapRecord<String, String, String> message){

        beforeHandleMessage(redisChannel,message);
        onMessage(redisChannel,message);
        afterHandleMessage(redisChannel,message);

    }


    public void beforeHandleMessage(IRedisChannel redisChannel,MapRecord<String, String, String> message){

    }

    protected void afterHandleMessage(IRedisChannel redisChannel,MapRecord<String, String, String> message){

        if (redisChannel.autoAck()) {
            redisChannel.ack();
        }

    }
    public abstract void onMessage(IRedisChannel redisChannel,MapRecord<String, String, String> message);


    @Override
    public T getObject() throws Exception {
            ProxyFactoryBean proxyFactoryBean=new ProxyFactoryBean();
            proxyFactoryBean.setAutodetectInterfaces(true);
            proxyFactoryBean.setBeanFactory(applicationContext);
            proxyFactoryBean.setTarget(this);
            proxyFactoryBean.setInterceptorNames(redisStreamInterceptors.stream()
                    .map(IRedisStreamInterceptor::getBeanName)
                    .toArray(String[]::new));
            proxyFactoryBean.setProxyTargetClass(true);
            return (T) proxyFactoryBean.getObject();
    }

    @Override
    public Class<?> getObjectType() {
        return type;
    }

    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }
}
